package org.logsin37.tool.eurekamanager.infra.repository.impl;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.logsin37.tool.eurekamanager.domain.entity.FrozenConfig;
import org.logsin37.tool.eurekamanager.infra.constant.ErrorMessages;
import org.logsin37.tool.eurekamanager.infra.repository.FrozenConfigRepository;
import org.springframework.data.redis.core.ReactiveHashOperations;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.stereotype.Repository;
import org.springframework.util.Assert;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

/**
 * Eureka RepositoryI mpl
 *
 * @author logsin37 2020/01/20 16:32
 */
@Repository
@Slf4j
@AllArgsConstructor
public class FrozenConfigRepositoryImpl implements FrozenConfigRepository {

    private final ReactiveStringRedisTemplate redisTemplate;
    private final ObjectMapper objectMapper;

    @Override
    public Flux<FrozenConfig> listFrozenConfig() {
        final ReactiveHashOperations<String, String, String> hashOperations = this.redisTemplate.opsForHash();
        return hashOperations.keys(CONFIG_KEY)
                .subscribeOn(Schedulers.elastic())
                .publishOn(Schedulers.elastic())
                .map(code -> new FrozenConfig().setEurekaCode(code));
    }

    @Override
    public Mono<FrozenConfig> getFrozenConfigByEurekaCode(String eurekaCode) {
        Assert.hasText(eurekaCode, String.format(ErrorMessages.NOT_NULL, "eureka code"));
        final ReactiveHashOperations<String, String, String> hashOperations = this.redisTemplate.opsForHash();
        return hashOperations.get(CONFIG_KEY, eurekaCode)
                .subscribeOn(Schedulers.elastic())
                .publishOn(Schedulers.elastic())
                .map(json -> {
                    try {
                        return this.objectMapper.readValue(json, FrozenConfig.class);
                    } catch (JsonProcessingException e) {
                        throw new RuntimeException(e.getMessage(), e);
                    }
                });
    }

    @Override
    public Mono<Void> saveFrozenConfig(FrozenConfig frozenConfig) {
        Assert.notNull(frozenConfig, String.format(ErrorMessages.NOT_NULL, "frozen config"));
        Assert.hasText(frozenConfig.getEurekaCode(), String.format(ErrorMessages.NOT_NULL, "eureka code"));
        return Mono.just(frozenConfig)
                .subscribeOn(Schedulers.elastic())
                .publishOn(Schedulers.elastic())
                .map(fc -> {
                    try {
                        return this.objectMapper.writeValueAsString(fc);
                    } catch (JsonProcessingException e) {
                        throw new RuntimeException(e.getMessage(), e);
                    }
                })
                .publishOn(Schedulers.elastic())
                .flatMap(json -> this.redisTemplate.opsForHash().put(CONFIG_KEY, frozenConfig.getEurekaCode(), json))
                .then();
    }

    @Override
    public Mono<Void> removeFrozenConfigByEurekaCode(String eurekaCode) {
        Assert.hasText(eurekaCode, String.format(ErrorMessages.NOT_NULL, "eureka code"));
        return this.redisTemplate.opsForHash().remove(CONFIG_KEY, eurekaCode)
                .subscribeOn(Schedulers.elastic())
                .publishOn(Schedulers.elastic())
                .then();
    }

    @Override
    public Mono<Void> saveFrozenRecord(String eurekaCode) {
        Assert.hasText(eurekaCode, String.format(ErrorMessages.NOT_NULL, "eureka code"));
        return Mono.just(eurekaCode)
                .subscribeOn(Schedulers.elastic())
                .publishOn(Schedulers.elastic())
                .flatMap(ec -> this.redisTemplate.opsForHash().put(STATUS_KEY, ec, String.valueOf(Boolean.TRUE)))
                .then();
    }

    @Override
    public Mono<Void> dropFrozenRecord(String eurekaCode) {
        Assert.hasText(eurekaCode, String.format(ErrorMessages.NOT_NULL, "eureka code"));
        return Mono.just(eurekaCode)
                .subscribeOn(Schedulers.elastic())
                .publishOn(Schedulers.elastic())
                .flatMap(ec -> this.redisTemplate.opsForHash().remove(STATUS_KEY, ec))
                .then();
    }

    @Override
    public Mono<Boolean> isFrozen(String eurekaCode) {
        Assert.hasText(eurekaCode, String.format(ErrorMessages.NOT_NULL, "eureka code"));
        return Mono.just(eurekaCode)
                .subscribeOn(Schedulers.elastic())
                .publishOn(Schedulers.elastic())
                .flatMap(ec -> this.redisTemplate.opsForHash().hasKey(STATUS_KEY, ec));
    }
}
